package cn.doitedu.hebust.pre

import java.text.SimpleDateFormat

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Dataset}

/**
  * @date: 2019/11/27
  * @site: www.doitedu.cn
  * @author: hunter.d 涛哥
  * @qq: 657270652
  * @description:
  * APP 埋点日志预处理
  */
object AppEventLogPreprocess {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)

    if(args.size<4){
      println(
        """
          |usage:
          |  参数1：原始日志输入路径
          |  参数2：地域字典输入路径
          |  参数3：处理结果输出路径
          |  参数4：运行模式(local/yarn)
        """.stripMargin)
      sys.exit(1)
    }

    // 输入日志数据的路径
    val srcdatapath = args(0)
    // 地域字典数据路径
    val areaDictPath = args(1)
    // 预处理输出路径
    val outputPath = args(2)
    // 运行模式
    val master = args(3)



    // 获取 sparksession对象
    val spark = SparkUtil.getSpark(this.getClass.getSimpleName,master)
    import spark.implicits._
    import spark.sql

    // 加载原始日志文件
    val ds: Dataset[String] = spark.read.textFile(srcdatapath)

    // 解析json
    val beans: Dataset[AppLogBean] = ds.map(AppLogBean.parseJson2Bean(_))


    /**
      * 清洗过滤
      */
    val validBeans: Dataset[AppLogBean] = beans
      .filter(_ != null)
      .filter(AppLogBean.isValidBean(_))


    /**
      * 数据集成（GPS坐标->省市区/商圈）
      */
    // 1. 先加载地理位置字典数据到Driver端，并广播
    val dictDF = spark.read.parquet(areaDictPath)
    val dictMap = DictLoadUtil.loadAreaDict(dictDF)
    val bc_dict = spark.sparkContext.broadcast(dictMap)

    // 2. 然后进行集成
    val integrated: Dataset[AppLogBean] = validBeans.mapPartitions(iter => {

      // 取广播变量
      val dict = bc_dict.value
      // 构造一个时间格式转化器
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

      // 遍历处理每一个bean
      iter.map(bean => {
        // 处理GPS坐标
        val longtitude = bean.longtitude
        val latitude = bean.latitude

        // 如果经纬度坐标在中国的经纬度范围之内，才去转geohash编码并从字典中查找省市区
        if (longtitude > 0 && longtitude < 120 && latitude > 0 && latitude < 70) {
          val geo = GeoHash.withCharacterPrecision(latitude, longtitude, 5).toBase32
          val area = dict.getOrElse(geo, ("未知省", "未知市", "未知区"))

          bean.province = area._1
          bean.city = area._2
          bean.district = area._3
        }

        // 处理时间戳
        val str = sdf.format(bean.timestamp).split(" ")
        bean.dateStr = str(0)
        bean.timeStr = str(1)

        // 返回集成完成的bean
        bean
      })
    })
    // integrated.where("trim(province) !='' ").show(10,false)

    /**
      * 数据修正（uid回补）
      */
    val haveUid = integrated.where(" uid is not null and trim(uid)!='' ")
    val noUid = integrated.where(" uid is null or trim(uid) ='' ")

    import org.apache.spark.sql.functions._
    val uids = haveUid
      .groupBy($"uid")
      .agg(
        max("imei").as("imei"),
        max("imsi").as("imsi"),
        max("mac").as("mac"),
        max("uuid").as("uuid"),
        max("androidId").as("androidId"),
        max("deviceId").as("deviceId")
      )

    noUid.createTempView("nouid")
    uids.createTempView("uids")

    // 开发一个自定义函数，用来判断两个字符串在非空情况下是否相等
    val is_equal = (x: String, y: String) => {
      if (x != y || StringUtils.isBlank(x) || StringUtils.isBlank(y))  false  else true
    }
    spark.udf.register("is_equal",is_equal)


    // 对没有uid的数据进行回补操作
    val part1: DataFrame = sql(
      """
        |
        |select
        |
        |a.eventid ,
        |a.timestamp ,
        |a.event ,
        |if(b.uid is not null,b.uid,a.uid) as uid,
        |a.phoneNbr ,
        |a.sessionId ,
        |a.imei ,
        |a.mac ,
        |a.imsi ,
        |a.osName ,
        |a.osVer ,
        |a.androidId ,
        |a.resolution ,
        |a.deviceType ,
        |a.deviceId ,
        |a.uuid ,
        |a.appid ,
        |a.appVer ,
        |a.release_ch ,
        |a.promotion_ch ,
        |a.longtitude ,
        |a.latitude ,
        |a.carrier ,
        |a.netType ,
        |a.cid_sn ,
        |a.ip ,
        |a.province,
        |a.city,
        |a.district,
        |a.dateStr,
        |a.timeStr
        |
        |from
        |
        |nouid  a left join uids  b
        |  on is_equal(a.imei,b.imei)
        |    or is_equal(a.imsi,b.imsi)
        |    or is_equal(a.mac,b.mac)
        |    or is_equal(a.uuid,b.uuid)
        |    or is_equal(a.androidId,b.androidId)
        |    or is_equal(a.deviceId,b.deviceId)
        |
      """.stripMargin)


    // 将回补好的数据  union  原来就有uid的数据
    val result = haveUid.toDF.union(part1)

    // 输出结果
    result.write.parquet(outputPath)

    spark.close()
  }

}
